package flink_p1

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
 * checkpoint测试
 */
object FlinkTest_14_CheckPoint {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /** checkpoint配置 * */

    //开启checkpoint
    env.enableCheckpointing(3000)
    //指定存储到hdfs
    env.setStateBackend(new FsStateBackend("file:///Users/sevenhong/code/flink_demo/ckdir"))
    //    env.setStateBackend(new FsStateBackend("hdfs://node1:9000/flink/ckdir14"))

    //    env.setStateBackend(new RocksDBStateBackend("hdfs://node1:9000/flink_rocksdb/ckdir14"))
    //默认 存储到JVM 堆内存中
    //    env.setStateBackend(new MemoryStateBackend(100 * 1024 * 1024))


    //checkpoint执行超时时间，默认10min
    env.getCheckpointConfig.setCheckpointTimeout(10 * 60 * 1000)

    //两次checkpoint的间隔，默认为-1，不使用这个，一旦配置了这个值，那么checkpoint就变成串行的了
    //    env.getCheckpointConfig.setCheckpointInterval(-1)

    //默认就是 EXACTLY_ONCE
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    //checkpoint最大并行度，默认：1
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(3)

    //默认：0  checkpoint 的最小间隔
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(600)


    /**
     * checkpoint允许失败的次数，超过则报错
     */
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(1)


    /**
     * 在取消任务时，处理处理checkpoint文件：
     * RETAIN_ON_CANCELLATION  : 保留
     * DELETE_ON_CANCELLATION : 删除
     */
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)


    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)


    socketStream
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)
      .print()


    env.execute()
  }

}
